package cn.doitedu.demo6;

import cn.doitedu.demo1.UserEvent_1;
import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.jcodings.util.Hash;

import java.util.HashMap;

public class Demo6 {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 配置一些必选参数
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");


        // 构建一个kafka source
        KafkaSource<String> source = KafkaSource.<String>builder()
                .setTopics("user-events-log")
                .setBootstrapServers("doitedu:9092")
                .setGroupId("g001")
                //.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                .setStartingOffsets(OffsetsInitializer.latest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 将source添加到env，得到数据流
        DataStreamSource<String> jsonStream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "s1");

        // json解析，封装成javaBean
        SingleOutputStreamOperator<UserEvent_1> beanStream = jsonStream.map(value -> JSON.parseObject(value, UserEvent_1.class));

        // 将数据keyBy(用户)
        KeyedStream<UserEvent_1, Long> keyByStream = beanStream.keyBy(UserEvent_1::getUid);

        // 处理
        SingleOutputStreamOperator<String> messageStream = keyByStream.process(new KeyedProcessFunction<Long, UserEvent_1, String>() {

            HashMap<String, RuleCalculator> ruleCalculatorHashMap = new HashMap<>();

            @Override
            public void open(Configuration parameters) throws Exception {
                RuntimeContext runtimeContext = getRuntimeContext();

                // 模型1的两个规则：
                String rule_1_1 = "{\"rule_id\":\"rule-1-001\",\"event_id\":\"X\"}";
                String rule_1_2 = "{\"rule_id\":\"rule-1-002\",\"event_id\":\"S\"}";

                // 模型2的两个规则：
                String rule_2_1 = "{\"rule_id\":\"rule-2-001\",\"profile_conditions\":[{\"tag_name\":\"age\",\"compare_type\":\"between\",\"condition_values\":[\"25\",\"40\"]},{\"tag_name\":\"gender\",\"compare_type\":\"=\",\"condition_values\":[\"female\"]},{\"tag_name\":\"active_level\",\"compare_type\":\">\",\"condition_values\":[\"2\"]}],\"realtime_profile\":{\"event_id\":\"C\",\"min_count\":3},\"fire_condition\":{\"event_id\":\"K\"}}";
                String rule_2_2 = "{\"rule_id\":\"rule-2-002\",\"profile_conditions\":[{\"tag_name\":\"gender\",\"compare_type\":\"=\",\"condition_values\":[\"female\"]}],\"realtime_profile\":{\"event_id\":\"P\",\"min_count\":2},\"fire_condition\":{\"event_id\":\"G\"}}";


                // 构造运算机对象
                RuleCalculator calculator1 = new RuleModel_1_Calculator();
                calculator1.init(runtimeContext,rule_1_1);
                ruleCalculatorHashMap.put("rule-1-001",calculator1);

                RuleCalculator calculator2 = new RuleModel_1_Calculator();
                calculator2.init(runtimeContext,rule_1_2);
                ruleCalculatorHashMap.put("rule-1-002",calculator2);


                // 构造模型2的两个规则运算机对象
                RuleCalculator calculator3 = new RuleModel_2_Calculator();
                calculator3.init(runtimeContext,rule_2_1);
                ruleCalculatorHashMap.put("rule-2-001",calculator3);

                RuleCalculator calculator4 = new RuleModel_2_Calculator();
                calculator4.init(runtimeContext,rule_2_2);
                ruleCalculatorHashMap.put("rule-2-002",calculator4);


            }

            @Override
            public void processElement(UserEvent_1 eventBean, KeyedProcessFunction<Long, UserEvent_1, String>.Context ctx, Collector<String> out) throws Exception {

                // 遍历运算机池
                // 调用运算机，处理当前收到的数据
                for (RuleCalculator ruleCalculator : ruleCalculatorHashMap.values()) {
                    ruleCalculator.calc(eventBean, out);
                }
            }
        });

        // 对外输出结果
        messageStream.print();

        env.execute();

    }

}
